package org.leo.dm.example.flow;

import java.util.HashMap;
import java.util.Map;

import org.leo.dm.common.bean.ResultBean;
import org.leo.dm.common.bean.StepResultBean;
import org.leo.dm.component.engine.DefaultStorage;
import org.leo.dm.component.flow.Flow;
import org.leo.dm.component.reader.db.DbReader;
import org.leo.dm.component.util.DmlEnum;
import org.leo.dm.example.base.FlowDaemon;
import org.leo.dm.example.base.db.TestDbSource;
import org.leo.dm.example.callback.CallBackExample;
import org.leo.dm.example.callback.DbExceptionCallbackImpl;
import org.leo.dm.example.plugin.PluginColumnCustom;
import org.leo.dm.example.plugin.PluginRecordCustom;
import org.leo.dm.example.writer.DbWriter;
import org.leo.dm.util.ThreadPoolUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class FlowDbToDbExample extends Flow {
	Logger log = LoggerFactory.getLogger(this.getClass());

	@Override
	public ResultBean handle(Map<String, Object> param) {
		ResultBean resultBean = new ResultBean();
		try {
			DbReader reader = new DbReader();
			reader.setConn(new TestDbSource().getConn());
			reader.setSourceTableNames(new String[] { "test3", "test" });
			reader.setColumnPlugin(new PluginColumnCustom());
			reader.setRecordPlugin(new PluginRecordCustom());
			reader.addExtendDataNames(new String[] { "GROUPID" });
			reader.setSqls(new String[] { "select * from test3 where rownum<30", "select * from test where rownum<20" });
			reader.setCallBack(new CallBackExample());

			DbWriter writer = new DbWriter();
			writer.setConn(new TestDbSource().getConn());

			Map<String, String> tableNameMapping = new HashMap<String, String>();
			tableNameMapping.put("test3", "test5");
			tableNameMapping.put("test", "test2");
			writer.setTableNameMapping(tableNameMapping);

			Map<String, Map<String, String>> tableColumnMapping = new HashMap<String, Map<String, String>>();
			Map<String, String> columnMapping = new HashMap<String, String>();
			// key为目标 value为原
			columnMapping.put("ID", "ID");
			columnMapping.put("A", "DATAGENERATE_DATE");
			columnMapping.put("B", "LOCAL_ID");
			columnMapping.put("C", "BUSINESS_ID");
			columnMapping.put("D", "BASIC_ACTIVE_ID");
			columnMapping.put("E", "RECORD_IDEN");
			columnMapping.put("GROUPID", "GROUPID");
			tableColumnMapping.put("test5", columnMapping);
			writer.setTableColumnMapping(tableColumnMapping);
			writer.setDmlEnum(DmlEnum.INSERT);
			writer.setCallback(new CallBackExample());
			writer.setDbExceptionCallback(new DbExceptionCallbackImpl());

			DefaultStorage storage = new DefaultStorage(20000);
			storage.setReader(reader);
			storage.setWriter(writer);

			this.setReader(reader);
			this.setWriter(writer);
			this.setStorage(storage);

			this.run();

			// 监控数据读写效率
			new FlowDaemon().handle(new ThreadPoolUtil(), reader, writer);

			// 同步等待返回值
			StepResultBean readerResultBean = reader.getResultBean();
			log.info("readerResultBean" + readerResultBean.toStr());
			StepResultBean writerResultBean = writer.getResultBean();
			log.info("writerResultBean" + writerResultBean.toStr());

			if ("-1".equals(readerResultBean.getCode()) || "-1".equals(writerResultBean.getCode())) {
				resultBean.setCode("-1");
			} else {
				resultBean.setCode("1");
			}

		} catch (Exception e) {
			e.printStackTrace();
			// 发生异常回滚已经插入的数据
			resultBean.setCode("-1");
		}

		return resultBean;
	}

	public static void main(String[] args) {
		final ThreadPoolUtil handlePoolUtil = new ThreadPoolUtil();
		for (int i = 0; i < 10; i++) {
			handlePoolUtil.getPool().execute(new Runnable() {
				@Override
				public void run() {
					Map<String, Object> param = new HashMap<String, Object>();
					FlowDbToDbExample flow = new FlowDbToDbExample();
					flow.handle(param);
				}
			});
		}

		handlePoolUtil.shutdownPool();
	}

}
